package mylab.io.core.context;

import com.google.common.collect.Maps;
import com.google.common.eventbus.Subscribe;

import java.util.Map;
import java.util.Objects;
import java.util.function.Consumer;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import mylab.base.foundation.lang.Callback;
import mylab.io.base.Das;
import mylab.io.base.context.DefaultDC;
import mylab.io.base.model.Endpoint;
import mylab.io.base.model.Msg;
import mylab.io.base.model.Terminal;
import mylab.io.base.property.DasProperties;
import mylab.io.base.util.TerminalUtil;
import mylab.io.core.cache.ChannelCache;
import mylab.io.core.cache.ChannelInfo;
import mylab.io.core.cache.DefaultChannelCache;
import mylab.io.core.event.ChannelActiveChangedEvent;
import mylab.io.core.event.ChannelExceptionEvent;
import mylab.io.core.event.ChannelIdleEvent;
import mylab.io.core.event.ChannelReadEvent;
import mylab.io.core.event.EndpointFoundEvent;
import mylab.utils.common.CastUtil;


@Slf4j
public class DefaultCC<P extends DasProperties, M extends Msg>
        extends DefaultDC<P, M> implements CoreContext<P, M> {

    protected Channel initChannel;
    protected ChannelCache channelCache;

    /**
     * 存储terminal对应的channel
     * <p>
     * 每个terminal对应唯一的channel,但channel可能对应多个terminal
     * <p>
     * eg:
     * tcp连接：一个NioSocketChannel对应一个对端
     * udp连接：一个NioDatagramChannel可以对应多个对端
     */
//    protected Map<Terminal, Channel> terminalChannelMap;
    protected Map<String, Channel> terminalChannelMap;

    public DefaultCC(P dasParams) {
        super(dasParams);
    }

    @Override
    public void init(@NonNull Das<?, P, M> das) {
        this.channelCache = new DefaultChannelCache();
        this.terminalChannelMap = Maps.newConcurrentMap();
        super.init(das);
    }

    @Override
    public void dispose() {
        this.channelCache = null;
        this.terminalChannelMap.clear();
        this.terminalChannelMap = null;

        super.dispose();
    }

    @Override
    public ChannelCache getChannelCache() {
        return channelCache;
    }

    @Override
    public void initChannel(Channel channel) {
        this.initChannel = channel;
    }


    @Override
    public Channel getChannel(Terminal terminal) {
        if (terminal == null) {
            return null;
        }

        terminalChannelMap.entrySet().forEach(entry -> log.info("terminalId:{}  channelId:{}", entry.getKey(), entry.getValue().id().asShortText()));

        if (terminalChannelMap.containsKey(terminal.getId())) {
            log.info("find channel. {} -> {}", terminal.getId(), terminalChannelMap.get(terminal.getId()).id().asShortText());
            return terminalChannelMap.get(terminal.getId());
        } else {
            return initChannel;
        }
    }

    //----------------------> channelEvent begin ------------------

    @Subscribe
    public void onChannelActiveEvent(ChannelActiveChangedEvent event) {
        String channelId = event.getSource().id().asShortText();
        boolean isActive = event.getContent();
        if (isActive) {
            ChannelInfo channelInfo = new ChannelInfo(channelId, das.getId());
            channelCache.updateChannel(channelInfo);
        } else {
            //移除terminal与channel的对应关系
            terminalChannelMap.forEach((terminal, channel) -> {
                if (Objects.equals(channelId, channel.id().asShortText())) {
                    terminalChannelMap.remove(terminal);
                }
            });

            //移除channel的内部的remotes
            if (channelCache.containsChannel(channelId)) {
                ChannelInfo channelInfo = channelCache.getChannel(channelId);
                channelInfo.getRemotes().values().forEach(endpoint -> onTerminalConnectionChanged(TerminalUtil.socketTerminal(endpoint), false));
            }

            //删除缓存中的channel
            channelCache.removeChannel(channelId);
        }
    }

    @Subscribe
    public void onEndpointFoundEvent(EndpointFoundEvent event) {
        Channel channel = event.getSource();
        String channelId = channel.id().asShortText();
        Endpoint endpoint = event.getContent();
        Terminal remote = TerminalUtil.socketTerminal(endpoint);
        remote.setDasId(das.getId());
        remote.setProtocol(getDasPropertites().getCommon().getProtocol());

        //记录terminal与channel的对应关系
        terminalChannelMap.put(remote.getId(), channel);

        //更新channel的内部的remotes
        ChannelInfo channelInfo = channelCache.getChannel(channelId);
        Map<String, Endpoint> remotes = channelInfo.getRemotes();
        if (!remotes.containsKey(remote.getId())) {
            remotes.put(remote.getId(), remote.getEndpoint());
        } else {
            remotes.replace(remote.getId(), remote.getEndpoint());
        }
        channelCache.updateChannel(channelInfo);

        onTerminalConnectionChanged(remote, true);
    }


    @Subscribe
    public void onChannelIdleEvent(ChannelIdleEvent event) {
        ChannelHandlerContext ctx = event.getSource();
        String dasId = das.getId();
        String channelId = ctx.channel().id().asShortText();

        //空闲时断开(可能是死连接),释放连接资源
        das.stop(Callback.simple((Consumer<Void>) unused -> log.warn("### onChannelIdleEvent ### 已经关闭通道!!! das id:{} channel id:{}", dasId, channelId)));
    }

    @Subscribe
    public void onChannelReadEvent(ChannelReadEvent event) {
        onReceived(CastUtil.cast(event.getContent()));
    }

    @Subscribe
    public void onChannelExceptionEvent(ChannelExceptionEvent event) {
        onException(event.getContent().getThrowable());
    }

    //----------------------> channelEvent end ------------------


}
